package com.lc1993929.dubbo.registry.consul;

import com.alibaba.dubbo.common.Constants;
import com.alibaba.dubbo.common.URL;
import com.alibaba.dubbo.common.logger.Logger;
import com.alibaba.dubbo.common.logger.LoggerFactory;
import com.alibaba.dubbo.common.utils.NamedThreadFactory;
import com.alibaba.dubbo.registry.NotifyListener;
import com.alibaba.dubbo.registry.support.FailbackRegistry;
import com.ecwid.consul.v1.*;
import com.ecwid.consul.v1.agent.model.NewService;
import com.ecwid.consul.v1.health.model.HealthService;
import com.google.common.collect.ImmutableMap;
import org.apache.http.client.HttpClient;
import org.apache.http.client.config.RequestConfig;
import org.apache.http.impl.client.HttpClientBuilder;
import org.apache.http.impl.conn.PoolingHttpClientConnectionManager;

import java.util.*;
import java.util.concurrent.*;
import java.util.concurrent.atomic.AtomicReference;

/**
 * Created by LiuChang on 2017/9/11/011.
 * consul的注册器，用于注册服务到consul和确认服务可用。类似于{@link com.alibaba.dubbo.registry.zookeeper.ZookeeperRegistry}
 */
public class ConsulRegistry extends FailbackRegistry {
    private static final Logger logger = LoggerFactory.getLogger(ConsulRegistry.class);

    private static final String TAG_DUBBO = "dubbo_service";
    private static final int DEFAULT_PORT = 8500;
    // httpClient配置
    private static final int DEFAULT_CONNECTION_TIMEOUT = 10000; // 10 sec
    private static final int DEFAULT_READ_TIMEOUT = 120000; // 120 sec for blocking Queries
    private static final int DEFAULT_WAIT_TIMEOUT = (DEFAULT_READ_TIMEOUT / 1000) - 20; // 100 sec default wait time for blocking Queries
    // second


    private ServiceIdGenerator serviceIdGenerator;

    private long ttl = 10000L;

    private ConsulClient consulClient;

    private final ConcurrentMap<URL, Watcher> notifiers = new ConcurrentHashMap<URL, Watcher>();
    private final ScheduledExecutorService keepAliveExecutor = Executors.newScheduledThreadPool(1, new NamedThreadFactory
            ("DubboConsulRegistryTimer", true));
    private final ScheduledFuture<?> keepAliveFuture;

    public ConsulRegistry(URL url, ServiceIdGenerator serviceIdGenerator) {
        super(url);
        if (url.isAnyHost()) {
            throw new IllegalStateException("consul address is not available");
        }
        String host = url.getHost();
        int port = url.getPort(DEFAULT_PORT);
        PoolingHttpClientConnectionManager connectionManager = new PoolingHttpClientConnectionManager();
        connectionManager.setMaxTotal(1000);
        connectionManager.setDefaultMaxPerRoute(500);

        RequestConfig requestConfig = RequestConfig.custom().
                setConnectTimeout(DEFAULT_CONNECTION_TIMEOUT).
                setConnectionRequestTimeout(DEFAULT_CONNECTION_TIMEOUT).
                setSocketTimeout(DEFAULT_READ_TIMEOUT).
                build();
        HttpClient httpClient = HttpClientBuilder.create().
                setConnectionManager(connectionManager).
                setDefaultRequestConfig(requestConfig).
                useSystemProperties().
                build();
        ConsulRawClient rawClient = new ConsulRawClient(host, port, httpClient);
        consulClient = new ConsulClient(rawClient);
        keepAliveFuture = keepAliveExecutor.scheduleWithFixedDelay(new Runnable() {
            public void run() {
                try {
                    keepAlive(); // 确保服务在consul的status为passing
                } catch (Throwable t) { //
                    logger.error("Unexpected exception occur at defer consul keep alive time, cause: " + t.getMessage(), t);
                }
            }
        }, ttl / 2, ttl / 2, TimeUnit.MILLISECONDS);
        if (serviceIdGenerator == null) {
            this.serviceIdGenerator = HasingIdGenerator.hasingIdGenerator;
        } else
            this.serviceIdGenerator = serviceIdGenerator;
    }

    @Override
    protected void doRegister(URL url) {
        NewService consulService = new NewService();
        consulService.setAddress(url.toFullString());
        consulService.setPort(url.getPort());
        consulService.setId(serviceIdGenerator.generate(url));
        consulService.setName(url.getServiceInterface());
        consulService.setCheck(buildHealthCheck(url));
        consulService.setTags(Arrays.asList(TAG_DUBBO));
        consulClient.agentServiceRegister(consulService);
    }

    @Override
    protected void doUnregister(URL url) {
        consulClient.agentServiceDeregister(serviceIdGenerator.generate(url));
    }

    /**
     * 订阅符合条件的已注册数据，当有注册数据变更时自动推送.
     * <p>
     * 订阅需处理契约：<br>
     * 1. 当URL设置了check=false时，订阅失败后不报错，在后台定时重试。<br>
     * 2. 当URL设置了category=routers，只通知指定分类的数据，多个分类用逗号分隔，并允许星号通配，表示订阅所有分类数据。<br>
     * 3. 允许以interface,group,version,classifier作为条件查询，如：interface=com.alibaba.foo.BarService&version=1.0.0<br>
     * 4. 并且查询条件允许星号通配，订阅所有接口的所有分组的所有版本，或：interface=*&group=*&version=*&classifier=*<br>
     * 5. 当注册中心重启，网络抖动，需自动恢复订阅请求。<br>
     * 6. 允许URI相同但参数不同的URL并存，不能覆盖。<br>
     * 7. 必须阻塞订阅过程，等第一次通知完后再返回。<br>
     *
     * @param url      订阅条件，不允许为空，如：consumer://10.20.153.10/com.alibaba.foo.BarService?version=1.0.0&application=kylin
     * @param listener 变更事件监听器，不允许为空
     */
    protected void doSubscribe(URL url, NotifyListener listener) {

        String serviceName = url.getServiceInterface();
        Watcher notifier = notifiers.get(url);
        if (notifier == null) {
            notifier = new Watcher(serviceName, url);
            notifiers.putIfAbsent(url, notifier);
            if (notifier == notifiers.get(url)) {
                notifier.start();
            }
        }
        if (Constants.ANY_VALUE.equals(url.getServiceInterface())) {
            com.ecwid.consul.v1.Response<Map<String, List<String>>> response = consulClient.getCatalogServices(null);
            Map<String, List<String>> catalogServices = response.getValue();
            List<URL> urlList = new ArrayList<URL>();
            for (String service : catalogServices.keySet()) {
                if (catalogServices.get(service).contains(TAG_DUBBO)) {
                    urlList.addAll(queryService(service));
                }
            }
            doNotify(url, urlList);

        } else {
            doNotify(url, queryService(serviceName));
        }
    }

    public List<URL> queryService(String serviceName) {
        com.ecwid.consul.v1.Response<List<HealthService>> healthServices = consulClient.getHealthServices(serviceName, true, null);
        List<URL> urlList = new ArrayList<URL>();
        for (HealthService service : healthServices.getValue()) {
            if (service.getService() == null || service.getService().getTags() == null || !service.getService().getTags().contains(TAG_DUBBO)) {
                logger.warn("ignore service [" + service.getService() + "]");
                continue;
            }
            urlList.add(URL.valueOf(service.getService().getAddress()));
        }
        return urlList;
    }

    @Override
    protected void doUnsubscribe(URL url, NotifyListener notifyListener) {
        notifiers.remove(url);
    }

    @Override
    public boolean isAvailable() {
        return consulClient.getAgentSelf().getValue() != null;
    }

    private void doNotify(URL url, List<URL> urlList) {
        for (NotifyListener listener : getSubscribed().get(url)) {
            doNotify(url, listener, urlList);
        }
    }

    @Override
    public void destroy() {
        super.destroy();
        for (Watcher notifier : notifiers.values()) {
            if (notifier != null) {
                notifier.shutdown();
                unregister(notifier.url);
            }
        }
    }

    /**
     * 创建health check , 使用TCP方式
     *
     * @param url
     * @return
     */
    private NewService.Check buildHealthCheck(URL url) {
        NewService.Check check = new NewService.Check();
        //        check.setTcp(url.getHost() + ":" + url.getPort());
        //        check.setInterval("10s");
        check.setTtl((ttl / 1000) + "s");
        check.setDeregisterCriticalServiceAfter("5m"); // 超过5m deregister service
        return check;
    }

    private void keepAlive() {
        for (URL url : new HashSet<URL>(getRegistered())) {
            if (url.getParameter(Constants.DYNAMIC_KEY, true)) {
                keepAlive(url);
            }
        }
    }

    public void keepAlive(URL url) {
        try {
            String checkId = serviceIdGenerator.generate(url);
            this.consulClient.agentCheckPass("service:" + checkId);
            logger.debug("Keep alive success, serviceId: " + checkId);
        } catch (OperationException e) {
            logger.error("Keep alive failure", e);
            if (e.getStatusCode() == 500) {
                logger.warn("Keep alive failure, Re-register service " + url.getServiceInterface());
                this.register(url);
            }
        }
    }

    private class Watcher extends Thread {
        private volatile boolean running = true;
        private String serviceName;
        private URL url;
        private final AtomicReference<ImmutableMap<URL, HealthService>> lastResponse = new AtomicReference<ImmutableMap<URL, HealthService>>
                (ImmutableMap.<URL, HealthService>of());
        private final AtomicReference<Long> lastIndex = new AtomicReference<Long>(Long.valueOf(0));

        public Watcher(String serviceName, URL url) {
            super("WatcherThread-" + serviceName);
            this.serviceName = serviceName;
            this.url = url;
        }

        @Override
        public void run() {
            while (this.running) {
                try {
                    ImmutableMap<URL, HealthService> full = doCheck();
                    if (!Objects.equals(full.keySet(), lastResponse.get().keySet())) {
                        lastResponse.set(full);
                        doNotify(url, full.keySet().asList());
                    }
                } catch (Throwable t) {
                    logger.warn("Unexpected Consul Exception:" + t.getMessage(), t);
                }
            }
        }

        public ImmutableMap<URL, HealthService> doCheck() {
            final ImmutableMap.Builder<URL, HealthService> builder = ImmutableMap.builder();
            Long last_index = lastIndex.get();
            Long _index;
            // Blocking Queries
            QueryParams queryParams = new QueryParams(DEFAULT_WAIT_TIMEOUT, last_index);
            if (Constants.ANY_VALUE.equals(url.getServiceInterface())) {
                Response<Map<String, List<String>>> catalogServices = consulClient.getCatalogServices(queryParams);
                _index = catalogServices.getConsulIndex();
                if (_index != null) {
                    lastIndex.set(_index);
                }
                if (_index <= last_index) {
                    return builder.build();
                }
                Set<String> serviceList = catalogServices.getValue().keySet();
                for (String name : serviceList) {
                    Response<List<HealthService>> healthServices = consulClient.getHealthServices(name, true, null);
                    for (HealthService service : healthServices.getValue()) {
                        if (service.getService() == null || service.getService().getTags() == null || !service.getService().getTags()
                                .contains(TAG_DUBBO)) {
                            logger.debug("ignore service [" + service.getService() + "]");
                            continue;
                        }
                        builder.put(URL.valueOf(service.getService().getAddress()), service);
                    }
                }


            } else {
                Response<List<HealthService>> healthServices = consulClient.getHealthServices(serviceName, true, queryParams);
                _index = healthServices.getConsulIndex();
                if (_index != null) {
                    lastIndex.set(_index);
                }
                if (_index <= last_index) {
                    return builder.build();
                }
                for (HealthService service : healthServices.getValue()) {
                    builder.put(URL.valueOf(service.getService().getAddress()), service);
                }

            }
            return builder.build();
        }

        public void shutdown() {
            this.running = false;
        }
    }

}
